package com.heima.article.stream;

import com.heima.common.constants.message.HotArticleConstants;
import com.heima.common.util.JsonUtils;
import com.heima.model.article.dtos.ArticleBehaviorMsg;
import com.heima.model.article.dtos.ArticleBehaviorStreamMsg;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;

@EnableBinding(BehaviorHandler.CountNumProcessor.class)
@Slf4j
public class BehaviorHandler {
    public interface CountNumProcessor {
        @Input(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC)
        KStream<String, String> input();

        @Output(HotArticleConstants.HOT_ARTICLE_SCORE_CONSUMER_TOPIC)
        KStream<String, String> output();
    }

    @StreamListener(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC)
    @SendTo(HotArticleConstants.HOT_ARTICLE_SCORE_CONSUMER_TOPIC)
    public KStream<String, String> process(KStream<String, String> stream) {
        return stream.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                log.info("收到了原始数据：{}", value);
                ArticleBehaviorMsg msg = JsonUtils.toBean(value, ArticleBehaviorMsg.class);
                return new KeyValue<>(msg.getArticleId().toString(), value);
            }
        }).groupByKey(Grouped.with(Serdes.String(), Serdes.String())).windowedBy(TimeWindows.of(Duration.ofSeconds(10)).grace(Duration.ZERO)).aggregate(new Initializer<String>() {
            @Override
            public String apply() {
                log.info("时间窗口初始化");
                return null;
            }
        }, new Aggregator<String, String, String>() {
            @Override
            public String apply(String key, String value, String aggregate) {
                ArticleBehaviorStreamMsg result = null;
                if (aggregate == null) {
                    // 时间窗口内的第一条消息
                    result = new ArticleBehaviorStreamMsg();
                    result.setArticleId(Long.parseLong(key));
                } else {
                    // 时间窗口内已经累加过消息了
                    result = JsonUtils.toBean(aggregate, ArticleBehaviorStreamMsg.class);
                }
                // 把当前的value累加到aggregate上去
                ArticleBehaviorMsg behaviorMsg = JsonUtils.toBean(value, ArticleBehaviorMsg.class);
                ArticleBehaviorMsg.ArticleBehaviorType msgType = behaviorMsg.getType();
                if (msgType == ArticleBehaviorMsg.ArticleBehaviorType.VIEWS) {
                    result.setViews(result.getViews() + behaviorMsg.getNum());
                } else if (msgType == ArticleBehaviorMsg.ArticleBehaviorType.LIKES) {
                    result.setLikes(result.getLikes() + behaviorMsg.getNum());
                } else if (msgType == ArticleBehaviorMsg.ArticleBehaviorType.COMMENT) {
                    result.setComment(result.getComment() + behaviorMsg.getNum());
                } else if (msgType == ArticleBehaviorMsg.ArticleBehaviorType.COLLECTION) {
                    result.setCollection(result.getCollection() + behaviorMsg.getNum());
                }
                log.info("时间窗口内的聚合计算，value：{}， aggregate：{}, new value:{}", value, aggregate, JsonUtils.toString(result));
                return JsonUtils.toString(result);
            }
        }, Materialized.with(Serdes.String(), Serdes.String())).suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded())).toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                return new KeyValue<>(key.key(), value);
            }
        });
    }
}
